package com.gpj.learn.rocketmq.transaction;

import com.gpj.learn.rocketmq.CommonConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;

/**
 * @className TransactionProducer
 * @description
 *   TODO
 * @author GPJ
 * @date 2021/4/26 23:16
 * @version 1.0
 **/
@Slf4j
public class TransactionProducer {
	public static int num = 0;
	public static void main(String[] args) {
		TransactionMQProducer producer = new TransactionMQProducer("TransactionMessageProducerGroup");
		producer.setNamesrvAddr(CommonConstants.ROCKETMQ_NAMESERVER_ADDR);

		producer.setExecutorService(new ThreadPoolExecutor(2, 5,
				100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000),
				runnable ->  new Thread(runnable, "message-check-thread") ));

		producer.setTransactionListener(new TransactionListener() {
			@Override
			public LocalTransactionState executeLocalTransaction(Message message, Object o) {
				if (StringUtils.equals(message.getTags(), "commit")) {
					log.info("消息在[事物提交]阶段执行[commit]:{}", message);
					return LocalTransactionState.COMMIT_MESSAGE;
				} else if (StringUtils.equals(message.getTags(), "fastfail")) {
					log.info("消息在[事物提交]阶段执行[roolback]:{}", message);
					return LocalTransactionState.ROLLBACK_MESSAGE;
				}else{
					log.info("消息在[事物提交]阶段执行[unknow]:{}", message);
					return LocalTransactionState.UNKNOW;
				}
			}

			@Override
			public LocalTransactionState checkLocalTransaction(MessageExt message) {
				if (StringUtils.equals(message.getTags(), "async")) {
					log.info("消息在[事物回查]阶段执行[commit]:{}", message);
					return LocalTransactionState.COMMIT_MESSAGE;
				} else if (StringUtils.equals(message.getTags(), "try")) {
					TransactionProducer.num++;
					if(TransactionProducer.num > 2){
						log.info("消息在第三次[事物回查]阶段执行[commit]:{}", message);
						return LocalTransactionState.COMMIT_MESSAGE;
					}
					return LocalTransactionState.UNKNOW;
				} else {
					log.info("消息在[事物回查]阶段执行[roolback]:{}", message);
					return LocalTransactionState.ROLLBACK_MESSAGE;
				}
			}
		});

		try {
			producer.start();

			producer.sendMessageInTransaction(new Message(CommonConstants.DEFAULT_TOPIC_NAME,
					"commit", "执行完事物就提交的消息".getBytes(StandardCharsets.UTF_8)), null);
			producer.sendMessageInTransaction(new Message(CommonConstants.DEFAULT_TOPIC_NAME,
					"fastfail", "执行事物失败的消息".getBytes(StandardCharsets.UTF_8)), null);
			producer.sendMessageInTransaction(new Message(CommonConstants.DEFAULT_TOPIC_NAME,
					"async", "回查事物的时候提交的消息".getBytes(StandardCharsets.UTF_8)), null);
			producer.sendMessageInTransaction(new Message(CommonConstants.DEFAULT_TOPIC_NAME,
					"try", "多次回查事物之后提交的消息".getBytes(StandardCharsets.UTF_8)), null);
			producer.sendMessageInTransaction(new Message(CommonConstants.DEFAULT_TOPIC_NAME,
					"fail", "回查事物失败的消息".getBytes(StandardCharsets.UTF_8)), null);
			log.info("消息全部发出");

			Thread.sleep(5 * 60 * 1000);
		} catch (MQClientException | InterruptedException e) {
			e.printStackTrace();
		}

	}
}
